package com.atguigu0.streaming

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket
import java.nio.charset.StandardCharsets

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver


/**
 * @description: xxx
 * @time: 2020/6/15 22:31
 * @author: baojinlong
 **/
class CustomerReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.MEMORY_ONLY) {
  // 启动时调用
  override def onStart(): Unit = {
    // 创建一个接收数据的线程
    new Thread("CustomerReceiver") {
      override def run(): Unit = {
        diyReceive()
      }
    }.start()
  }

  // 接收数据
  def diyReceive(): Unit = {
    var socket: Socket = null
    var reader: BufferedReader = null
    try {
      socket = new Socket(host, port)
      // 获取输入流
      reader = new BufferedReader(new InputStreamReader(socket.getInputStream, StandardCharsets.UTF_8))
      // 获取一行数据
      var input: String = reader.readLine
      while (input != null) {
        store(input)
        input = reader.readLine
      }
    } catch {
      case e: Exception =>
        println("发生了异常", e)
        reader.close()
        socket.close()
        restart("正在重启")
    }
  }

  /**
   * 停止时候调用
   */
  override def onStop(): Unit = {
    println("我发生了异常,正在停止.")
  }
}
